package com.example;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @Author: wyp
 * @Date: 2021/9/13 21:43
 */
public class BlockingQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        MyMessageQueue<String> myMessageQueue = new MyMessageQueue<>(new ArrayBlockingQueue<>(10));

        new Thread(() -> {
            try {
                myMessageQueue.producer("生产的消息");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "生产线程A").start();

        Thread.sleep(10000);

        new Thread(() -> {
            try {
                myMessageQueue.consumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println();
        }, "消费线程B").start();

        //线程睡眠10秒
        Thread.sleep(10000);

        myMessageQueue.stop();
        System.out.println(Thread.currentThread().getName() + "\t 发送了stop() 指令！");
    }
}

/**
 * 简单版消息队列
 */
class MyMessageQueue<T> {

    /**
     * 默认开启消息队列
     */
    private volatile boolean flag = true;

    private BlockingQueue<T> blockingQueue = null;

    public MyMessageQueue(BlockingQueue<T> blockingQueue) {
        this.blockingQueue = blockingQueue;
        System.out.println(Thread.currentThread().getName() + "\t 当前Blocking：" + blockingQueue.getClass().getSimpleName());
    }

    public void producer(T message) throws Exception {
        boolean offer;
        while (flag) {
            offer = blockingQueue.offer(message, 2L, TimeUnit.SECONDS);
            if (offer) {
                System.out.println(Thread.currentThread().getName() + "\t 消息推送成功，消息：" + message.toString());
            } else {
                System.out.println(Thread.currentThread().getName() + "\t消息推送失败，消息：" + message.toString());
            }
            Thread.sleep(2000);
        }

        System.out.println(Thread.currentThread().getName() + "\t消息队列生产者即将关闭！");
    }

    public void consumer() throws Exception {
        T poll;
        while (flag) {
            poll = blockingQueue.poll(2L, TimeUnit.SECONDS);
            if (poll != null) {
                System.out.println(Thread.currentThread().getName() + "\t消息队列消费到：" + poll);
            } else {
                System.out.println(Thread.currentThread().getName() + "\t消息队列获取为空！");
            }
        }

        System.out.println(Thread.currentThread().getName() + "\t消息队列消费者即将关闭！");
    }

    public void stop() {
        this.flag = false;
    }
}
